package com.bestcem.xm.common.core.config.delaymq.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.date.DateField;
import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.bestcem.xm.common.core.config.delaymq.dao.MqSendLogDao;
import com.bestcem.xm.common.core.config.delaymq.entity.MqSendLogDO;
import com.bestcem.xm.common.core.config.delaymq.enums.MqSendStatusEnum;
import com.bestcem.xm.common.core.config.delaymq.service.DelayMessageService;
import com.bestcem.xm.common.core.config.delaymq.util.DelayMqUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.data.redis.core.DefaultTypedTuple;
import org.springframework.data.redis.core.ZSetOperations;
import org.springframework.http.MediaType;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.*;
import java.util.stream.Collectors;

/**
 * @author sh.pei <sh.pei@idiaoyan.com>
 * @version v1.0
 * @date 2022/1/13 16:19
 */
@Component
@Slf4j
public class DelayMessageServiceImpl implements DelayMessageService {

    @Resource
    private StreamBridge bridge;

    @Resource
    private DelayMqUtil delayMqUtil;

    @Resource(name = "mqDelayMqExecutor")
    private ThreadPoolTaskExecutor mqDelayMqExecutor;

    /**
     * 消息批处理数量
     */
    private static final int batchCount = 100;

    /**
     * 日期在 maxDayGap 之内，则立即将消息投递入redis
     */
    private static final int MAX_DAY_GAP = 7;

    @Resource
    private MqSendLogDao mqSendLogDao;

    /**
     * 单次查询最大值,做限制
     */
    private static final int batchQueryCount = 10000;

    /**
     * 从redis的zset里得到要到期的mq消息,并将消息投递到mq里。
     */
    @Override
    public void pickMsgFromRedis() {
        int queueSize = mqDelayMqExecutor.getThreadPoolExecutor().getQueue().size();
        if (queueSize > 0) {
            if (log.isInfoEnabled()) {
                log.info("[delayMq] delayMq ThreadPool queue not empty, queueSize:{} , return directly. wait for next fire", queueSize);
            }
            return;
        }
        Date now = com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate();
        double score = delayMqUtil.getScore(now);
        Set<String> keys = delayMqUtil.rangeByScore(Integer.MIN_VALUE, score);
        if (CollUtil.isEmpty(keys)) {
            return;
        } else {
            if (log.isInfoEnabled()) {
                log.info("[delayMq] pickMsgFromRedis 待发送消息数量:{} 待发送消息数量:{}", keys.size(), JSON.toJSONString(keys));
            }
        }
        List<String> list = new ArrayList<>(keys.size());
        list.addAll(keys);
        int start = 0;
        do {
            mqDelayMqExecutor.execute(new MqSendRunnable(list.subList(start, Math.min(start += batchCount, list.size()))));
        } while (start < list.size());

    }

    /**
     * 从数据库里查到所有待加入到redis的消息,加入redis;
     */
    @Override
    public void pickToToAddRedisMsgFromDataBase() {
        /*ZSetOperations.TypedTuple<String> objectTypedTuple1 = new DefaultTypedTuple<>("eee",9.6);
        ZSetOperations.TypedTuple<String> objectTypedTuple2 = new DefaultTypedTuple<>("fff",1.5);
        ZSetOperations.TypedTuple<String> objectTypedTuple3 = new DefaultTypedTuple<>("ggg",7.4);
        Set<ZSetOperations.TypedTuple<String>> typles = new HashSet<>();
        typles.add(objectTypedTuple1);
        typles.add(objectTypedTuple2);
        typles.add(objectTypedTuple3);
        redisTemplate.opsForZSet().add("zSet", typles);*/

        Date now = com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate();
        // 从数据库里查到所有待加入到redis的消息,加入redis;
        List<MqSendLogDO> mqSendLogDOS = mqSendLogDao.listAwardMqSendLogs(MqSendStatusEnum.TO_ADD_REDIS.getIndex(), null, DateUtil.offset(now, DateField.DAY_OF_MONTH, MAX_DAY_GAP).toJdkDate(), batchQueryCount);
        if (CollUtil.isNotEmpty(mqSendLogDOS)) {
            Set<ZSetOperations.TypedTuple<Object>> tuples = mqSendLogDOS.stream().map(mqSendLogDO -> new DefaultTypedTuple<Object>(mqSendLogDO.getId(), delayMqUtil.getScore(mqSendLogDO.getDelayTime()))).collect(Collectors.toSet());
            /*Set<ZSetOperations.TypedTuple<Object>> tuples = new HashSet<>();
            for (MqSendLogDO item : mqSendLogDOS) {
                DefaultTypedTuple<Object> tuple = new DefaultTypedTuple<>(item.getId(), delayMqUtil.getScore(item.getDelayTime()));
                tuples.add(tuple);
            }*/
            delayMqUtil.addToRedis(tuples);
            batchUpdateStatus(mqSendLogDOS, 0, MqSendStatusEnum.TO_SEND.getIndex(), now);
        } else {
            log.info("[delayMq] pickToSendMsgFromDataBase 没有消息待发送");
        }
    }

    /**
     * 从数据库里查到所有超时但未从redis发送出去的消息;正常情况不会存在这种消息,如果存在这种消息,邮件告警处理
     */
    @Override
    public void pickToToSendMsgFromDataBase() {
        Date now = com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate();
        // 从数据库里查到所有超时但未从redis发送出去的消息
        List<MqSendLogDO> mqSendLogDOS = mqSendLogDao.listAwardMqSendLogs(MqSendStatusEnum.TO_SEND.getIndex(), null, DateUtil.offset(now, DateField.MINUTE, -1).toJdkDate(), batchQueryCount);
        if (CollUtil.isNotEmpty(mqSendLogDOS)) {
            // 邮件告警
            List<String> ids = mqSendLogDOS.stream().map(MqSendLogDO::getId).collect(Collectors.toList());
            log.error("[delayMq] 从数据库里查到所有超时但未从redis发送出去的消息,{}", ids);
            // 修改状态,防止重复发送
            batchUpdateStatus(mqSendLogDOS, 0, MqSendStatusEnum.REDIS_FAIL.getIndex(), now);
        } else {
            log.info("[delayMq] pickToToSendMsgFromDataBase 没有消息待发送");
        }
    }


    /**
     * 批量更新消息状态
     *
     * @param mqSendLogDOS 消息列表
     * @param newStatus    目标状态
     * @param updateTime   更新时间
     */
    private void batchUpdateStatus(List<MqSendLogDO> mqSendLogDOS, int countModify, int newStatus, Date updateTime) {
        if (CollUtil.isNotEmpty(mqSendLogDOS)) {
            List<String> ids = mqSendLogDOS.stream().map(MqSendLogDO::getId).collect(Collectors.toList());
            mqSendLogDao.batchUpdateStatus(ids, countModify, newStatus, updateTime);
        }

    }

    /**
     * 从数据库查到所有发送失败的消息,进行补发;
     */
    @Override
    public void pickSendFailMsgFromDataBase() {
        Date now = com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate();
        // 从数据库里查到所有待加入到redis的消息,加入redis;
        List<MqSendLogDO> mqSendLogDOS = mqSendLogDao.listAwardMqSendLogs(MqSendStatusEnum.SEND_FAIL.getIndex(), null, null, batchQueryCount);
        if (CollUtil.isNotEmpty(mqSendLogDOS)) {
            Set<ZSetOperations.TypedTuple<Object>> tuples = mqSendLogDOS.stream().map(mqSendLogDO -> new DefaultTypedTuple<Object>(mqSendLogDO.getId(), delayMqUtil.getScore(mqSendLogDO.getDelayTime()))).collect(Collectors.toSet());
            /*Set<ZSetOperations.TypedTuple<Object>> tuples = new HashSet<>();
            for (MqSendLogDO item : mqSendLogDOS) {
                DefaultTypedTuple<Object> tuple = new DefaultTypedTuple<>(item.getId(), delayMqUtil.getScore(item.getDelayTime()));
                tuples.add(tuple);
            }*/
            delayMqUtil.addToRedis(tuples);
            batchUpdateStatus(mqSendLogDOS, 0, MqSendStatusEnum.TO_SEND.getIndex(), now);
        } else {
            log.info("[delayMq] pickSendFailMsgFromDataBase 没有消息待发送");
        }
    }

    /**
     * 删除大于1月的消息;
     */
    @Override
    public void deleteUselessMsg() {
        Date date = DateUtil.offset(com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate(), DateField.MONTH, -1).toJdkDate();
        mqSendLogDao.deleteMqSendLogs(Arrays.asList(MqSendStatusEnum.SEND_SUCCESS.getIndex(), MqSendStatusEnum.UTTER_FAIL.getIndex()), date);
    }

    class MqSendRunnable implements Runnable {

        private final List<String> ids;

        MqSendRunnable(List<String> ids) {
            this.ids = ids;
        }

        @Override
        public void run() {
            List<MqSendLogDO> mqSendLogDOS = mqSendLogDao.listByIds(ids);
            if (CollUtil.isNotEmpty(mqSendLogDOS)) {
                Date now = com.bestcem.xm.common.core.uitls.DateUtil.getCommonDate();
                List<MqSendLogDO> sendSuccessList = new LinkedList<>();
                List<MqSendLogDO> sendFailList = new LinkedList<>();
                List<MqSendLogDO> utterFailList = new LinkedList<>();
                List<MqSendLogDO> ignoreList = new LinkedList<>();
                mqSendLogDOS.forEach(mqSendLogDO -> {
                    try {
                        long removeCount = delayMqUtil.removeId(mqSendLogDO.getId());
                        if (removeCount > 0) {
                            bridge.send(mqSendLogDO.getBindingName(), Base64.getDecoder().decode(mqSendLogDO.getSendMsg()), MediaType.APPLICATION_OCTET_STREAM);
                            sendSuccessList.add(mqSendLogDO);
                        } else {
                            ignoreList.add(mqSendLogDO);
                        }
                    } catch (Exception e) {
                        log.error("[delayMq] 延迟消息发送失败，id:{}", mqSendLogDO.getId());
                        // 判断重试次数,超过 彻底失败
                        int count = mqSendLogDO.getCount() == null ? 0 : mqSendLogDO.getCount();
                        if (count < delayMqUtil.getDelayRetryCount()) {
                            sendFailList.add(mqSendLogDO);
                        } else {
                            utterFailList.add(mqSendLogDO);
                        }
                    }
                });
                // 批量更新状态
                batchUpdateStatus(sendSuccessList, 1, MqSendStatusEnum.SEND_SUCCESS.getIndex(), now);
                batchUpdateStatus(sendFailList, 1, MqSendStatusEnum.SEND_FAIL.getIndex(), now);
                batchUpdateStatus(utterFailList, 1, MqSendStatusEnum.UTTER_FAIL.getIndex(), now);
                if (log.isInfoEnabled()) {
                    log.info("[delayMq] MqSendRunnable, sumCount:{}, success count:{} failCount:{}, utterFailCount:{}, ignoreCount:{}", ids.size(), sendSuccessList.size(), sendFailList.size(), utterFailList.size(), ignoreList.size());
                }
                // 错误次数大于阈值,发送告警邮件
                if (utterFailList.size() > 0) {
                    List<String> utterFailIds = utterFailList.stream().map(MqSendLogDO::getId).collect(Collectors.toList());
                    log.error("[delayMq] 延迟消息多次发送失败:{}", utterFailIds);
                }
            }
        }
    }
}
